use std::collections::VecDeque;
use std::sync::Arc;

use bytes::Bytes;
use session::{ClientSession, ServerSession};
use upstream::{HttpUpstreamCtl, HttpUpstreamStatus};
use variable::{HttpVariableTable, GLOBAL_HTTP_VARIABLE_TABLE};

use crate::config::http::site_conf::HttpAddressSiteRuntimeConf;
use crate::error::*;
use crate::modules::ModuleMgr;
use crate::{alog_debug, utils::freelock::FreeLock};

use super::base::headers::HttpHeaders;
use super::base::parse::HttpResult;
use super::pool::client_pool::ClientConnPools;
use super::{
    base::{
        conn::BoxHttpConn, header_name::HeaderName, status_code::StatusCode,
        Response,
    },
    BoxError, Error, ErrorType,
};

pub mod session;
pub mod upstream;
pub mod variable;

#[allow(dead_code)]
pub struct HttpProxyWrap {
    // 使用FreeLock包裹HttpProxy
    // 允许同一时间可以获取到多个HttpProxy的可变所有权.
    inner: FreeLock<HttpProxy>,
}

#[allow(dead_code)]
pub struct HttpProxy {
    server_session: ServerSession,
    variables: HttpVariableTable,
    conf: Option<Arc<HttpAddressSiteRuntimeConf>>,
    client_session: Option<ClientSession>,
    // 性能排查
    client_hc_key: Option<String>,

    downstream_resp: Response,
    upstream_ctl: HttpUpstreamCtl,
    pub exit_status: u16,
    pub exit_body: Option<Bytes>,
    error: Option<BoxError>,
    logged: bool,
}

#[allow(dead_code)]
impl HttpProxy {
    fn new(server_session: ServerSession) -> Self {
        HttpProxy {
            server_session,
            variables: GLOBAL_HTTP_VARIABLE_TABLE.read().copy(),
            conf: None,
            client_session: None,
            client_hc_key: None,
            downstream_resp: Response::new(),
            upstream_ctl: HttpUpstreamCtl::default(),
            exit_status: 0,
            exit_body: None,
            error: None,
            logged: false,
        }
    }

    #[inline]
    fn set_exit_status(&mut self, code: u16) {
        if self.exit_status != 0 {
            return;
        }
        self.exit_status = code;
    }

    #[inline]
    fn set_error(&mut self, err: BoxError) {
        if self.error.is_some() {
            return;
        }
        self.error = Some(err);
    }

    // 读取request body发送给上游.
    async fn proxy_to_upstream(&mut self) -> Result<()> {
        let client_session = self.client_session.as_mut().unwrap();
        let mut bufs = VecDeque::with_capacity(4);

        let mut done = false;
        while !done {
            let res = self.server_session.read_request_body(&mut bufs).await?;
            done = res == HttpResult::Complete;
            client_session.write_req_body(&mut bufs, done).await?;
        }

        Ok(())
    }

    // 读取response 发送给下游.
    async fn proxy_to_downstream(&mut self) -> Result<()> {
        let client_session = self.client_session.as_mut().unwrap();
        let res = client_session.read_response().await?.clone();
        self.downstream_resp.set(res);

        self.header_filter().await?;

        let mut bufs = VecDeque::with_capacity(4);

        let mut done = false;
        while !done {
            let client_session = self.client_session.as_mut().unwrap();
            let res = client_session.read_resp_body(&mut bufs).await?;
            done = res == HttpResult::Complete;
            self.body_filter(&mut bufs, done).await?;
        }

        let client_session = self.client_session.as_mut().unwrap();
        let trailers = client_session.get_resp_trailer_headers()?;

        if let Some(trailers) = trailers {
            let trailers_clone = trailers.clone();
            self.trailers_filter(&trailers_clone).await?;
        }

        Ok(())
    }

    async fn response_error(&mut self) -> Result<()> {
        if self.exit_status == 0 {
            self.exit_status = StatusCode::HTTP_INTERNAL_SERVER_ERROR;
        }

        if self.downstream_resp.status_code == 0 {
            self.downstream_resp.status_code = self.exit_status;
        }

        let mut cl = 0;
        if let Some(body) = &self.exit_body {
            cl = body.len();
        }

        self.downstream_resp.headers.set_header(
            Bytes::from_static(HeaderName::CONTENT_LENGTH),
            Bytes::copy_from_slice(cl.to_string().as_bytes()),
        );

        self.header_filter().await?;

        let mut bufs = VecDeque::new();
        if let Some(exit_body) = self.exit_body.as_ref() {
            bufs.push_back(exit_body.clone());
        }
        self.body_filter(&mut bufs, true).await?;

        // There is no need to send trailers in case of a response error.
        // self.trailers_filter().await?;

        Ok(())
    }

    async fn access_phase(&mut self) {
        if let Err(e) = ModuleMgr::http_access_handles(self).await {
            if self.error.is_none() {
                self.error = Some(e);
            }

            if self.exit_status == 0 {
                self.exit_status = StatusCode::HTTP_INTERNAL_SERVER_ERROR;
            }
        }
    }

    pub async fn header_final_filter(&mut self) -> Result<()> {
        if self.server_session.header_send() {
            return Error::e_explain(
                ErrorType::InternalError,
                "response header already send",
            );
        }
        self.server_session
            .send_resp_header(&mut self.downstream_resp)
            .await
    }

    async fn header_filter(&mut self) -> Result<()> {
        ModuleMgr::process_http_header_filter(self).await
    }

    pub async fn body_final_filter(
        &mut self,
        bufs: &mut VecDeque<Bytes>,
        done: bool,
    ) -> Result<HttpResult> {
        self.server_session.write_resp_body(bufs, done).await
    }

    async fn body_filter(
        &mut self,
        bufs: &mut VecDeque<Bytes>,
        done: bool,
    ) -> Result<HttpResult> {
        ModuleMgr::process_http_body_filter(self, bufs, done).await
    }

    async fn trailers_filter(
        &mut self,
        trailer_headers: &HttpHeaders,
    ) -> Result<()> {
        if self.server_session.expect_write_trailers_num() == 0 {
            return Ok(());
        }

        self.server_session
            .write_resp_trailers(trailer_headers)
            .await
    }

    async fn log_phase(&mut self) {
        // access log
    }
}

#[allow(dead_code)]
impl HttpProxyWrap {
    pub fn new(hc: BoxHttpConn) -> Self {
        let server_session = match hc.raw_conn {
            super::base::conn::HttpStream::H2Stream(_) => {
                ServerSession::new_http1(hc)
            }
            _ => ServerSession::new_http1(hc),
        };

        let proxy = HttpProxy::new(server_session);

        HttpProxyWrap {
            inner: FreeLock::new(proxy),
        }
    }

    pub async fn process_new(mut self) -> Option<BoxHttpConn> {
        let mut inner = self.inner.lock();

        let requests = inner.server_session.get_hc().requests;
        let _ = match inner.server_session.read_request().await {
            Ok(Some(req)) => req,
            Ok(None) => {
                alog_debug!(
                    "wait request timeout or client close, requests: {}",
                    requests
                );
                return self.finalize().await;
            }
            Err(e) => {
                inner.exit_status = StatusCode::HTTP_BAD_REQUEST;
                inner.error = Some(e);
                return self.finalize().await;
            }
        };

        let conf = match &inner.server_session.get_hc().address_site_conf_info {
            Some(info) => {
                if !info.valid {
                    inner.exit_status = StatusCode::HTTP_FORBIDDEN;
                    return self.finalize().await;
                }

                info.conf.clone()
            }

            None => {
                inner.exit_status = StatusCode::HTTP_FORBIDDEN;
                return self.finalize().await;
            }
        };

        let next_upstream_tries = conf.proxy_runtime_conf.next_backend_tries;

        inner.conf = Some(conf);

        // process access phase.
        inner.access_phase().await;
        if inner.exit_status != 0 {
            return self.finalize().await;
        }

        let mut tries = 0;

        loop {
            let upstream_status = HttpUpstreamStatus::default();
            self.inner
                .lock()
                .upstream_ctl
                .status_vec
                .push(upstream_status);

            let e = match self.do_proxy().await {
                Ok(()) => {
                    break;
                }

                Err(e) => e,
            };

            let mut inner = self.inner.lock();
            let last_status = inner.upstream_ctl.get_last_status();
            last_status.error = Some(e);

            // 重试次数达到上限.
            if tries > next_upstream_tries + 1 {
                break;
            }

            tries += 1;

            if !inner.allow_retry() {
                break;
            }
        }

        let mut inner = self.inner.lock();
        let last_status = inner.upstream_ctl.get_last_status();
        if let Some(e) = &last_status.error {
            let msg = e.to_string();
            inner.error = Some(Error::from_msg(e.etype.clone(), msg));
            if inner.exit_status == 0 {
                inner.exit_status = StatusCode::HTTP_BAD_GATEWAY;
            }
            return self.finalize().await;
        }

        self.finalize().await
    }

    async fn do_proxy(&mut self) -> Result<()> {
        let mut inner = self.inner.lock();

        let client_session = inner.connect_to_upstream().await?;
        inner.client_session = Some(client_session);

        // 先发送request headers.
        let mut client_req = inner.server_session.request().clone();
        inner
            .client_session
            .as_mut()
            .unwrap()
            .send_req_header(&mut client_req)
            .await?;

        let mut inner_upstream = self.inner.lock();
        let mut inner_downstream = self.inner.lock();

        tokio::try_join!(
            inner_upstream.proxy_to_upstream(),
            inner_downstream.proxy_to_downstream(),
        )?;

        // FIXME: 细节需要进一步设计.
        let client_session = inner.client_session.take();
        if let Some(mut client_session) = client_session {
            client_session.finalize(false).await;
            let client_hc = client_session.release();
            if let Some(client_hc) = client_hc {
                ClientConnPools::put_conn(
                    inner.client_hc_key.as_ref().unwrap(),
                    client_hc,
                    10000,
                );
            }
        }

        Ok(())
    }

    async fn finalize(self) -> Option<BoxHttpConn> {
        let mut inner = self.inner.lock();

        let header_send = inner.server_session.header_send();
        if !header_send {
            if let Err(e) = inner.response_error().await {
                if inner.error.is_none() {
                    inner.error = Some(e);
                }
            }
        }

        let abort = inner.error.is_some();
        inner.server_session.finalize(abort).await;

        if !inner.logged {
            inner.logged = true;
            inner.log_phase().await;
        }

        let inner = self.inner.into_inner();
        inner.server_session.release()
    }
}
